package org.argus.job.server.system.management.service;

import com.baomidou.mybatisplus.core.conditions.query.LambdaQueryWrapper;
import com.baomidou.mybatisplus.core.conditions.update.LambdaUpdateWrapper;
import com.baomidou.mybatisplus.core.toolkit.Wrappers;
import com.baomidou.mybatisplus.extension.plugins.pagination.Page;
import com.fasterxml.jackson.core.JsonProcessingException;
import lombok.extern.slf4j.Slf4j;
import org.apache.commons.lang3.StringUtils;
import org.argus.common.core.enums.StatusEnum;
import org.argus.common.core.exception.ServiceException;
import org.argus.common.core.utils.JsonUtils;
import org.argus.common.web.core.snowflake.SnowflakeIdGenerator;
import org.argus.job.common.SJ;
import org.argus.job.common.enums.InstanceStatus;
import org.argus.job.common.enums.TimeExpressionType;
import org.argus.job.common.model.AlarmConfig;
import org.argus.job.common.model.JobInfo;
import org.argus.job.common.model.LifeCycle;
import org.argus.job.common.model.LogConfig;
import org.argus.job.common.request.http.SaveJobInfoRequest;
import org.argus.job.server.system.dispatch.DispatchService;
import org.argus.job.server.system.domain.InstanceInfoEntity;
import org.argus.job.server.system.domain.JobInfoEntity;
import org.argus.job.server.system.instance.InstanceService;
import org.argus.job.server.system.management.mapper.InstanceInfoMapper;
import org.argus.job.server.system.management.mapper.JobInfoMapper;
import org.argus.job.server.system.management.model.JobInfoSearchReq;
import org.springframework.beans.BeanUtils;
import org.springframework.beans.factory.annotation.Autowired;
import org.springframework.stereotype.Service;
import org.argus.job.server.system.scheduler.TimingStrategyService;
import org.springframework.transaction.annotation.Transactional;
import org.springframework.util.CollectionUtils;
import tech.powerjob.server.common.timewheel.holder.InstanceTimeWheelService;

import java.util.ArrayList;
import java.util.Date;
import java.util.List;
import java.util.Optional;
import java.util.stream.Collectors;

/**
 * @Author 刘文/Cliff Liu
 * @Date 2024/8/7
 */
@Service
@Slf4j
public class JobInfoService {

    @Autowired
    private JobInfoMapper mapper;

    @Autowired
    private TimingStrategyService timingStrategyService;

    @Autowired
    private InstanceService instanceService;

    @Autowired
    private DispatchService dispatchService;

    @Autowired
    private InstanceInfoMapper instanceInfoMapper;

    @Autowired
    private SnowflakeIdGenerator idGenerator;

    public Page<JobInfo> page(Page page, JobInfoSearchReq pageReq) {
        LambdaQueryWrapper<JobInfoEntity> queryWrapper = Wrappers.lambdaQuery(JobInfoEntity.class);
        if (pageReq.getAppId() != null) {
            queryWrapper.eq(JobInfoEntity::getAppId, pageReq.getAppId());
        }

        if (StringUtils.isNotBlank(pageReq.getJobName())) {
            queryWrapper.like(JobInfoEntity::getJobName, pageReq.getJobName());
        }

        if (StringUtils.isNotBlank(pageReq.getJobDescription())) {
            queryWrapper.like(JobInfoEntity::getJobDescription, pageReq.getJobDescription());
        }

        if (StringUtils.isNotBlank(pageReq.getStatus())) {
            queryWrapper.eq(JobInfoEntity::getStatus, pageReq.getStatus());
        }

        Page result = mapper.selectPage(page, queryWrapper);
        List<JobInfo> records = new ArrayList<>(result.getRecords().size());
        for(Object o : result.getRecords()){
            JobInfoEntity jobInfoEntity = (JobInfoEntity) o;
            try {
                LifeCycle lifecycle = JsonUtils.parseObject(jobInfoEntity.getLifecycle(), LifeCycle.class);
                jobInfoEntity.setLifecycleRange(lifecycle.toLifecycleRange());
            } catch (JsonProcessingException e) {
                log.error("parse lifecycle error, job id" + jobInfoEntity.getId(), e);
            }
            JobInfo jobInfo = new JobInfo();
            BeanUtils.copyProperties(jobInfoEntity, jobInfo);
            if (StringUtils.isNotBlank(jobInfoEntity.getLogConfig())) {
                jobInfo.setLogConfig(JsonUtils.parseObjectIgnoreException(jobInfoEntity.getLogConfig(), LogConfig.class));
            }
            records.add(jobInfo);
        }
        result.setRecords(records);
        return result;
    }

    /**
     * 复制任务
     *
     * @param jobId 目标任务ID
     * @return 复制后的任务 ID
     */
    public JobInfoEntity copyJob(Long jobId) {

        JobInfoEntity origin = mapper.selectById(jobId);
        if (origin == null || origin.getStatus() == StatusEnum.DELETED) {
            throw new IllegalStateException("can't copy the job which has been deleted!");
        }
        JobInfoEntity copyJob = new JobInfoEntity();
        // 值拷贝
        BeanUtils.copyProperties(origin, copyJob);
        // 填充默认值，理论上应该不需要
        fillDefaultValue(copyJob);
        // 修正创建时间以及更新时间
        copyJob.setId(idGenerator.nextId());
        copyJob.setJobName(copyJob.getJobName() + "_COPY");

        mapper.insert(copyJob);
        return copyJob;

    }

    /**
     * 保存/修改任务
     *
     * @param request 任务请求
     * @return 创建的任务ID（jobId）
     */
    public Long saveJob(SaveJobInfoRequest request) {

        request.valid();
        validate(request);

        JobInfoEntity jobInfo;
        if (request.getId() != null) {
            jobInfo = mapper.selectById(request.getId());
            if (jobInfo == null) {
                throw new ServiceException("Job not exist");
            }
        } else {
            jobInfo = new JobInfoEntity();
        }

        // 值拷贝
        BeanUtils.copyProperties(request, jobInfo);

        // 填充默认值，非空保护防止 NPE
        fillDefaultValue(jobInfo);

        // 转化报警用户列表
        if (request.getNotifyUserIds() != null) {
            if (request.getNotifyUserIds().size() == 0) {
                jobInfo.setNotifyUserIds(null);
            } else {
                jobInfo.setNotifyUserIds(SJ.COMMA_JOINER.join(request.getNotifyUserIds()));
            }
        }
        LifeCycle lifecycle = LifeCycle.of(request.getLifecycleRange());
        jobInfo.setLifecycle(JsonUtils.toJSONString(lifecycle));
        // 检查定时策略
        timingStrategyService.validate(request.getTimeExpressionType(), request.getTimeExpression(),
            lifecycle.getStart(), lifecycle.getEnd());
        calculateNextTriggerTime(jobInfo);
        if (request.getId() == null) {
            jobInfo.setCreateTime(new Date());
        }

        // 检查告警配置
        if (request.getAlarmConfig() != null) {
            AlarmConfig config = request.getAlarmConfig();
            if (config.getStatisticWindowLen() == null || config.getAlertThreshold() == null
                || config.getSilenceWindowLen() == null) {
                throw new ServiceException("illegal alarm config!");
            }
//            jobInfo.setAlarmConfig(JSON.toJSONString(request.getAlarmConfig()));
        }
        // 日志配置
        if (request.getLogConfig() != null) {
            jobInfo.setLogConfig(JsonUtils.toJSONString(request.getLogConfig()));
        }
        // 日志配置
        if (request.getAdvancedRuntimeConfig() != null) {
            jobInfo.setAdvancedRuntimeConfig(JsonUtils.toJSONString(request.getAdvancedRuntimeConfig()));
        }
        if (jobInfo.getId() == null) {
            jobInfo.setId(idGenerator.nextId());
            mapper.insert(jobInfo);
        } else {
            mapper.updateById(jobInfo);
        }
        return jobInfo.getId();
    }

    public void validate(SaveJobInfoRequest saveJobInfoRequest) {
        LambdaQueryWrapper<JobInfoEntity> queryWrapper =
            Wrappers.lambdaQuery(JobInfoEntity.class).eq(JobInfoEntity::getJobName, saveJobInfoRequest.getJobName());
        if (saveJobInfoRequest.getId() != null) {
            queryWrapper.ne(JobInfoEntity::getId, saveJobInfoRequest.getId());
        }
        if (mapper.selectCount(queryWrapper) > 0) {
            throw new ServiceException("Same name job already exists");
        }

        timingStrategyService.validate(saveJobInfoRequest.getTimeExpressionType(),
            saveJobInfoRequest.getTimeExpression(), null, null);
    }

    /**
     * 手动立即运行某个任务
     *
     * @param jobId 任务ID
     * @param instanceParams 任务实例参数（仅 OpenAPI 存在）
     * @param delay 延迟时间，单位 毫秒
     * @return 任务实例ID
     */
    @Transactional(rollbackFor = Exception.class)
    public long runJob(Long appId, Long jobId, String instanceParams, Long delay) {

        delay = delay == null ? 0 : delay;
        JobInfoEntity jobInfo = mapper.selectById(jobId);
        if (jobInfo == null) {
            throw new IllegalArgumentException("can't find job by id:" + jobId);
        }

        log.info("[Job-{}] try to run job in app[{}], instanceParams={},delay={} ms.", jobInfo.getId(), appId,
            instanceParams, delay);
        final InstanceInfoEntity instanceInfo = instanceService.create(jobInfo.getId(), jobInfo.getAppId(),
            jobInfo.getJobParams(), instanceParams, null, System.currentTimeMillis() + Math.max(delay, 0));
        if (delay <= 0) {
            dispatchService.dispatch(jobInfo, instanceInfo.getId(), Optional.of(instanceInfo), Optional.empty());
        } else {
            InstanceTimeWheelService.schedule(instanceInfo.getId(), delay,
                () -> dispatchService.dispatch(jobInfo, instanceInfo.getId(), Optional.empty(), Optional.empty()));
        }
        log.info("[Job-{}|{}] execute 'runJob' successfully, params={}", jobInfo.getId(), instanceInfo.getId(),
            instanceParams);
        return instanceInfo.getId();
    }

    private void calculateNextTriggerTime(JobInfoEntity jobInfo) {
        // 计算下次调度时间
        if (TimeExpressionType.FREQUENT_TYPES.contains(jobInfo.getTimeExpressionType())) {
            // 固定频率类型的任务不计算
            jobInfo.setNextTriggerTime(null);
        } else {
            LifeCycle lifeCycle = LifeCycle.parse(jobInfo.getLifecycle());
            Long nextValidTime = timingStrategyService.calculateNextTriggerTimeWithInspection(
                jobInfo.getTimeExpressionType(), jobInfo.getTimeExpression(), lifeCycle.getStart(), lifeCycle.getEnd());
            jobInfo.setNextTriggerTime(nextValidTime);
        }
        // 重写最后修改时间
        jobInfo.setUpdateTime(new Date());
    }

    private void fillDefaultValue(JobInfoEntity jobInfo) {
        if (jobInfo.getMaxWorkerCount() == null) {
            jobInfo.setMaxWorkerCount(0);
        }
        if (jobInfo.getMaxInstanceNum() == null) {
            jobInfo.setMaxInstanceNum(0);
        }
        if (jobInfo.getConcurrency() == null) {
            jobInfo.setConcurrency(5);
        }
        if (jobInfo.getInstanceRetryNum() == null) {
            jobInfo.setInstanceRetryNum(0);
        }
        if (jobInfo.getTaskRetryNum() == null) {
            jobInfo.setTaskRetryNum(0);
        }
        if (jobInfo.getInstanceTimeLimit() == null) {
            jobInfo.setInstanceTimeLimit(0L);
        }
    }

    public SaveJobInfoRequest exportJob(Long jobId) {
        JobInfoEntity jobInfo = mapper.selectById(jobId);
        if (jobInfo == null) {
            throw new IllegalArgumentException("can't find job by jobId: " + jobId);
        }
        final SaveJobInfoRequest saveJobInfoRequest = JobConverter.convertJobInfoEntity2SaveJobInfoRequest(jobInfo);
        saveJobInfoRequest.setId(null);
        saveJobInfoRequest.setJobName(saveJobInfoRequest.getJobName() + "_EXPORT_" + System.currentTimeMillis());
        log.info("[Job-{}] [exportJob] jobInfoEntity: {}, saveJobInfoRequest: {}", jobId, JsonUtils.toJSONString(jobInfo), JsonUtils.toJSONString(saveJobInfoRequest));
        return saveJobInfoRequest;
    }

    public JobInfoEntity fetchJob(Long jobId) {
        return mapper.selectById(jobId);
    }

    public List<JobInfoEntity> fetchAllJob(Long appId) {
        return mapper.selectList(Wrappers.lambdaQuery(JobInfoEntity.class).eq(JobInfoEntity::getAppId, appId));
    }



    /**
     * 删除某个任务
     *
     * @param jobId 任务ID
     */
    public void deleteJob(Long jobId) {
        shutdownOrStopJob(jobId, StatusEnum.DELETED);
    }

    public void disableJob(Long jobId) {
        shutdownOrStopJob(jobId, StatusEnum.INACTIVE);
    }

    /**
     * 启用某个任务
     *
     * @param jobId 任务ID
     */
    public void enableJob(Long jobId) {
        JobInfoEntity jobInfoEntity = mapper.selectById(jobId);
        if(jobInfoEntity == null){
            new IllegalArgumentException("can't find job by jobId:" + jobId);
        }

        jobInfoEntity.setStatus(StatusEnum.NORMAL);
        calculateNextTriggerTime(jobInfoEntity);

        mapper.updateById(jobInfoEntity);
    }

    /**
     * 停止或删除某个JOB
     * 秒级任务还要额外停止正在运行的任务实例
     */
    private void shutdownOrStopJob(Long jobId, StatusEnum status) {

        // 1. 先更新 job_info 表
        JobInfoEntity jobInfoEntity = mapper.selectById(jobId);
        if (jobInfoEntity == null) {
            throw new IllegalArgumentException("can't find job by jobId:" + jobId);
        }
        jobInfoEntity.setStatus(status);
        mapper.updateById(jobInfoEntity);

        // 2. 关闭秒级任务
        if (!TimeExpressionType.FREQUENT_TYPES.contains(jobInfoEntity.getTimeExpressionType())) {
            return;
        }
        LambdaQueryWrapper<InstanceInfoEntity> queryWrapper = Wrappers.lambdaQuery(InstanceInfoEntity.class)
            .eq(InstanceInfoEntity::getJobId, jobId).in(InstanceInfoEntity::getStatus, InstanceStatus.RUNNING);
        List<InstanceInfoEntity> executeLogs = instanceInfoMapper.selectList(queryWrapper);
        if (CollectionUtils.isEmpty(executeLogs)) {
            return;
        }
        if (executeLogs.size() > 1) {
            log.warn("[Job-{}] frequent job should just have one running instance, there must have some bug.", jobId);
        }
        executeLogs.forEach(instance -> {
            try {
                // 重复查询了数据库，不过问题不大，这个调用量很小
                instanceService.stopInstance(instance.getId());
            } catch (Exception ignore) {
                // ignore exception
            }
        });
    }

    public List<JobInfo> list(JobInfoSearchReq searchReq) {
        LambdaQueryWrapper<JobInfoEntity> queryWrapper =  Wrappers.lambdaQuery(JobInfoEntity.class).ne(JobInfoEntity::getStatus, StatusEnum.DELETED);
        if (searchReq.getAppId() != null) {
            queryWrapper.eq(JobInfoEntity::getAppId, searchReq.getAppId());
        }
        return mapper.selectList(queryWrapper).stream().map(entity -> {
            JobInfo jobInfo = new JobInfo();
            BeanUtils.copyProperties(entity, jobInfo);
            return jobInfo;
        }).collect(Collectors.toList());
    }

    public void updateStatus(JobInfo jobInfo) {
        LambdaUpdateWrapper<JobInfoEntity> updateWrapper = Wrappers.lambdaUpdate(JobInfoEntity.class)
                .set(JobInfoEntity::getStatus, jobInfo.getStatus()).eq(JobInfoEntity::getId, jobInfo.getId());
        mapper.update(null, updateWrapper);
    }
}
